package com.cantor.consumer.discovery.impl;

import cn.hutool.core.net.url.UrlBuilder;
import com.cantor.common.exception.FailToConnectProviderException;
import com.cantor.common.util.CantorUtil;
import com.cantor.consumer.discovery.ServiceDiscoverer;
import com.cantor.consumer.start.ConsumerNettyKeeper;
import com.cantor.core.center.RegistrationCenter;
import com.cantor.core.pool.CantorExecutorPool;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 服务节点发现者之---不对nodes进行缓存, 实时查询zookeeper节点信息
 */
@Slf4j
public class NoCacheServiceDiscovererImpl implements ServiceDiscoverer {

    @Override
    @SneakyThrows
    public List<String> discover(RegistrationCenter center, String serviceNameWithVersion) {
        // 从center中得到这个服务的所有provider注册的节点信息, 然后缓存到自己的map中, 并监听断开事件.
        List<String> nodes = center.getChildren(serviceNameWithVersion);
        // 准备用于缓存的List
        List<String> cacheList = new CopyOnWriteArrayList<>();
        // 遍历这些节点, 通知ConnectionsKeeper建立连接, 直到建立成功后, 这个方法再return.
        CountDownLatch latch = new CountDownLatch(nodes.size());
        nodes.forEach(node -> {
            CompletableFuture.runAsync(()->{
                UrlBuilder urlBuilder = CantorUtil.getUrlBuilder(node);
                if (ConsumerNettyKeeper.addConnection(urlBuilder.getHost(), urlBuilder.getPort())) {
                    latch.countDown();
                    cacheList.add(node);
                }
            }, CantorExecutorPool.pool());
        });
        // 如果有部分Provider连接没成功,抛异常
        latch.await(5, TimeUnit.SECONDS);
        if (latch.getCount() != 0) {
            throw new FailToConnectProviderException();
        }
        return cacheList;
    }

}
